package influxdb

import (
	"context"
	"encoding/json"
	"fmt"
	db2 "github.com/influxdata/influxdb-client-go/v2"
	"github.com/influxdata/influxdb-client-go/v2/api"
	"regexp"
	"time"
)

const (
	//org = "glue"
	org    = "prometheus"
	bucket = "data"
)

type Client struct {
	db     db2.Client
	query  api.QueryAPI
	write  api.WriteAPI
	delete api.DeleteAPI

	db2     db2.Client
	query2  api.QueryAPI
	write2  api.WriteAPI
	delete2 api.DeleteAPI
}

func OpenInfluxdb(url, token, url2, token2 string) (*Client, error) {
	db := db2.NewClient(url, token)
	db.Options().HTTPClient().Timeout = time.Minute
	q := db.QueryAPI(org)
	w := db.WriteAPI(org, bucket)
	d := db.DeleteAPI()

	db2 := db2.NewClient(url2, token2)
	db2.Options().HTTPClient().Timeout = time.Minute
	q2 := db2.QueryAPI(org)
	w2 := db2.WriteAPI(org, bucket)
	d2 := db2.DeleteAPI()

	return &Client{
		db:     db,
		query:  q,
		write:  w,
		delete: d,

		db2:     db2,
		query2:  q2,
		write2:  w2,
		delete2: d2,
	}, nil
}

func (r *Client) PutLogs(data interface{}) error {
	d := data.(map[string]interface{})
	message := d["message"]
	//fmt.Printf("\n%v\n", d)
	t, ok := d["timestamp"].(float64)
	var tm time.Time
	if !ok {
		tm = time.Now()
	} else {
		tm = time.Unix(int64(t), 0)
	}
	delete(d, "message")
	delete(d, "timestamp")
	saveData := make(map[string]string, len(d))
	for k, v := range d {
		if vs, ok := v.(string); !ok {
			return fmt.Errorf("not %v string \n%v\n", v, d)
		} else {
			saveData[k] = vs
		}
	}

	p := db2.NewPoint("logs",
		saveData,
		map[string]interface{}{
			"message": message,
		},
		tm,
	)

	r.write2.WritePoint(p)
	r.write2.Flush()
	return nil
}

func (r *Client) Putk8sLogs(data interface{}) error {
	//w := r.impl.WriteAPI("vector", "logs")
	d := data.(map[string]interface{})
	//fmt.Printf("\n\n data: %v \n\n", d)

	k8s, ok := d["kubernetes"]
	if ok {
		delete(d, "kubernetes")
		k8sm := k8s.(map[string]interface{})
		for k, v := range k8sm {
			d[k] = v
		}
	}

	saveData := make(map[string]string, len(d))
	var message string
	for k, v := range d {
		val, ok := v.(string)
		if ok {
			saveData[k] = val
			if k == "message" {
				message = val
				delete(d, k)
				continue
			}
		} else {
			bts, _ := json.Marshal(v)
			saveData[k] = string(bts)
		}
	}
	timestamp := saveData["timestamp"]

	//file := d["file"].(string)
	//message := d["message"].(string)
	//typ := d["source_type"].(string)
	//stream := d["stream"].(string)
	//timestamp := d["timestamp"].(string)

	//kd := d["kubernetes"].(map[string]interface{})
	//k8sCid := kd["container_id"].(string)
	//k8sImg := kd["container_image"].(string)
	//k8sName := kd["container_name"].(string)
	//K8sIp := kd["pod_ip"].(string)
	//
	//k8sPodName := kd["pod_name"].(string)
	//k8sPodNameSpace := kd["pod_namespace"].(string)
	//k8sPodNodeName := kd["pod_node_name"].(string)
	//k8sPodUid := kd["pod_uid"].(string)
	//
	//namespaceLabels, _ := json.Marshal(kd["namespace_labels"])
	//podAnnotations, _ := json.Marshal(kd["pod_annotations"])
	//k8sIps, _ := json.Marshal(kd["pod_ips"])
	//podLabels, _ := json.Marshal(kd["pod_labels"])

	//guid := k8sPodNameSpace
	T, err := time.Parse(time.RFC3339Nano, timestamp)
	if err != nil {
		return err
	}

	var logType = "info"
	if ok, _ := regexp.MatchString("(?i)warn", message); ok {
		logType = "warn"
	} else if ok, _ := regexp.MatchString("(?i)error", message); ok {
		logType = "error"
	}

	saveData["logtype"] = logType

	p := db2.NewPoint("logs",
		saveData,
		map[string]interface{}{
			"message": message,
		},
		T,
	)

	r.write2.WritePoint(p)

	//r.write.WriteRecord(fmt.Sprintf("data,guid=%s,node=%s,type=%s,ip=%s,cid=%s,img=%s,pod_name=%s,container_name=%s, "+
	//	"file=%s,message=%s,stream=%s,"+
	//	"pod_uid=%s,namespace=%s,namespaces_labels=%v,"+
	//	"pod_annotations=%v,pod_ips=%v,pod_labels=%v,"+
	//	" %v",
	//	guid, k8sPodNodeName, typ, K8sIp, k8sCid, k8sImg, k8sPodName, k8sName,
	//	file, message, stream,
	//	k8sPodUid, k8sPodNameSpace, namespaceLabels,
	//	podAnnotations, k8sIps, podLabels,
	//	timestamp))

	r.write2.Flush()
	return nil
}

func (r *Client) GetList(ctx context.Context, sql string, data interface{}) error {
	res, err := r.query.Query(ctx, sql)
	if err != nil {
		return err
	}

	Res := []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		Res = append(Res, map[string]interface{}{
			"label": d["_field"],
			"time":  d["_time"],
			"value": d["_value"],
		})
	}
	//fmt.Printf("%v", Res)
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetLogsInfo(ctx context.Context, sql string, data interface{}) error {
	res, err := r.query2.Query(ctx, sql)
	if err != nil {
		return err
	}

	Res := []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		Res = append(Res, map[string]interface{}{
			"container_id": d["cid"],
			"ip":           d["ip"],
			"type":         d["type"],
			"logtype":      d["logtype"],
			"start":        d["_start"],
			"stop":         d["_stop"],
			"value":        d["_value"],
		})
	}
	//fmt.Printf("%v", Res)
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetLogs(ctx context.Context, sql string, data interface{}) error {

	res, err := r.query2.Query(ctx, sql)
	//fmt.Printf("\n\n %v %v \n\n", sql, res)
	if err != nil {
		return err
	}

	Res := []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		Res = append(Res, d)
	}
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetStatus(ctx context.Context, sql string, data interface{}) error {
	res, err := r.query.Query(ctx, sql)
	if err != nil {
		return err
	}

	Res := []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		Res = append(Res, d)
	}
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetMetrics(ctx context.Context, sql string, data interface{}) error {
	//fmt.Printf("sql: %v", sql)

	res, err := r.query.Query(ctx, sql)
	if err != nil {
		return err
	}

	Res := []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		Res = append(Res, d)
	}
	//fmt.Printf("%v", Res)
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetK8sNodes(ctx context.Context, sql string, data interface{}) error {
	res, err := r.query.Query(ctx, sql)
	if err != nil {
		return err
	}

	var Res = []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		//Res[d["instance"].(string)] = append(Res[d["instance"].(string)], map[string]interface{}{
		//	"time": d["_time"].(time.Time).Local(),
		//	"node": d["node"],
		//})
		Res = append(Res, map[string]interface{}{
			"node": d["node"].(string),
		})
	}
	//fmt.Printf("%v", Res)
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetK8sPodContainers(ctx context.Context, sql string, data interface{}) error {
	res, err := r.query.Query(ctx, sql)
	if err != nil {
		return err
	}

	var Res = []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		Res = append(Res, d)
	}
	//fmt.Printf("%v", Res)
	*data.(*[]map[string]interface{}) = Res
	return nil
}

func (r *Client) GetK8sContainerInfo(ctx context.Context, sql string, data interface{}) error {
	res, err := r.query.Query(ctx, sql)
	if err != nil {
		return err
	}

	var Res = []map[string]interface{}{}
	for res.Next() {
		d := res.Record().Values()
		delete(d, "table")
		Res = append(Res, d)
	}
	//fmt.Printf("%v", Res)
	*data.(*[]map[string]interface{}) = Res
	return nil
}
